package Autodesk;

import java.io.IOException;
import java.util.Hashtable;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Reducer.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import Autodesk.TotalVolumeFilter.LogMapper;
import Autodesk.TotalVolumeFilter.LogReducer;

public class FeatureFilter extends Configured implements Tool {
	public Path prodPath;
	public static class LogMapper extends Mapper<Object, Text, Text, IntWritable> {
		
		private final static IntWritable one = new IntWritable(1);
		private Text word = new Text();
		Hashtable<String, String> ht = new Hashtable<String, String>();

	public void map(Object key, Text value, Context context)
			throws IOException, InterruptedException {
		StringBuffer bf = new StringBuffer();

		String record[] = value.toString().toLowerCase().split("\t");
		try{
		bf.append(record[0]);// Date.
		bf.append("\t");
		bf.append(record[1]);// Channel
		bf.append("\t");
		for(int i=9;i<15;i++){//Feature
			word.set(bf.toString()+record[i]);
			context.write(word, one);
		}
		}catch(Exception e){
			System.out.println("Erroor Reg: "+value);
		}
	}
}
public static class Combine extends
Reducer<Text, IntWritable, Text, IntWritable> {
	private IntWritable result = new IntWritable();

	@Override
	public void reduce(Text key, Iterable<IntWritable> values,
			Context context) throws IOException, InterruptedException {
		int sum = 0;
		for (IntWritable val : values) {
			sum += val.get();
		}
		result.set(sum);
		context.write(key, result);		
	}
	
}
public static class LogReducer extends
		Reducer<Text, IntWritable, Text, IntWritable> {
	
	private IntWritable result = new IntWritable();

	public void reduce(Text key, Iterable<IntWritable> values,
			Context context) throws IOException, InterruptedException {
		
		int sum = 0;
		for (IntWritable val : values) {
			sum += val.get();
		}
		result.set(sum);
		context.write(key, result);
	
	}
}

public int run(String[] args) throws Exception { 
	Configuration conf = getConf();
	
	Job job = new Job(conf, "ParseLogs");
	job.setJarByClass(Autodesk.TotalVolumeFilter.class);

	job.setMapperClass(LogMapper.class);
	job.setReducerClass(LogReducer.class);
	job.setCombinerClass(Combine.class);
	
	job.setOutputKeyClass(Text.class);
	job.setOutputValueClass(IntWritable.class);

    FileInputFormat.setInputPaths(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]) );

	System.exit(job.waitForCompletion(true)?0:1);
	
	return 0;

}

public static void main(String[] args) throws Exception {
	int res = ToolRunner.run(new Configuration(), new FeatureFilter(), args);
	System.exit(res);
	}
}
